package cn.itcast.czxy.BD18

import cn.itcast.czxy.BD18.bean.BaseMode
import org.apache.spark.sql.DataFrame

object AgeGroup extends BaseMode{
  override def setAppName: String = "AnageGroup"

  override def setLeven4Id: Int = 117

  override def getNewTag(leve5: DataFrame, hbaseDF: DataFrame): DataFrame = {
    import spark.implicits._
    val wjbq = leve5.map(row => {
      val id = row.getAs("id").toString
      val rule: Array[String] = row.getAs("rule").toString.split("-")
      (id, rule(0), rule(1))
    }).toDF("id", "start", "stop")

    val usersj = hbaseDF.map(row => {
      val id: String = row.getAs("id").toString
      val birthdays: Array[String] = row.getAs("birthday").toString.split("-")
      (id, birthdays(0))
    }).toDF("id", "birthday")
    usersj.join(wjbq,usersj("birthday").between(wjbq("start"),wjbq("stop"))).select(usersj("id").as("userId"),wjbq("id").as("tagsId"))
  }

  def main(args: Array[String]): Unit = {
    exec()
  }
}
